package com.zk.storm;
//每日设备最大点击量
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

public class DailyDeviceMaxBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Map<String, Integer> deviceMaxMap;  // 键：date#device，值：最大点击量

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.deviceMaxMap = new HashMap<>();
    }

    @Override
    public void execute(Tuple input) {
        try {
            String dateKey = input.getStringByField("date_key");
            String device = input.getStringByField("device");
            int hits = input.getIntegerByField("hits");

            String key = dateKey + "#" + device;
            int currentMax = deviceMaxMap.getOrDefault(key, 0);
            if (hits > currentMax) {
                deviceMaxMap.put(key, hits);
            }

            collector.emit(new Values(key, deviceMaxMap.get(key)));
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
            throw new RuntimeException("设备最大点击量聚合失败: " + e.getMessage());
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("agg_key", "max_hits"));  // agg_key=date#device
    }
}